package com.bjbn.service;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.List;

@Service
public class KafkaProducerService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 发送单条消息
     */
    public void sendMessage(String topic, Object message) {
        kafkaTemplate.send(topic, message);
    }

    /**
     * 批量发送列表数据
     */
    public void sendBatchMessages(String topic, List<?> messages) {
        messages.forEach(message -> kafkaTemplate.send(topic, message));
    }

    /**
     * 批量发送带回调
     */
    public void sendBatchMessagesWithCallback(String topic, List<?> messages) {
        messages.forEach(message -> {
            kafkaTemplate.send(topic, message).addCallback(
                    result -> {
                        if (result != null) {
                            System.out.printf("Sent message=[%s] with offset=[%d] to partition=[%d]%n",
                                    message, result.getRecordMetadata().offset(), result.getRecordMetadata().partition());
                        }
                    },
                    ex -> System.err.println("Unable to send message=[" + message + "] due to : " + ex.getMessage())
            );
        });
    }
}